/*
 * This file is part of the <a href="http://code.google.com/p/pux">PUX Framework</a>
 *
 * Copyright 2008 by <a href="mailto:jfendler@polytechnic.edu.na">Jens Fendler</a>
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *
 */

package de.teamskill.pux.engine;

import java.util.Collection;
import java.util.Map;
import java.util.TreeMap;

/**
 * AbstractNode
 * 
 * @author <a href="mailto:jfendler@polytechnic.edu.na">Jens Fendler</a>
 * 
 */
public abstract class AbstractNode implements Runnable, Logging {

	/**
	 * NODE_READY_WAIT_DELAY
	 */
	public static final long NODE_READY_WAIT_DELAY = 50;

	/**
	 * autoNodeNr
	 */
	private static int autoNodeNr = 1;

	/**
	 * nr - number of this node object (used by the default constructor)
	 */
	protected static int nr = 1;

	/**
	 * inputPortMap
	 */
	private Map<String, InputPort> inputPortMap = new TreeMap<String, InputPort>();

	/**
	 * outputPortMap
	 */
	private Map<String, OutputPort> outputPortMap = new TreeMap<String, OutputPort>();

	/**
	 * running: true, while the node's thread is running, false otherwise
	 */
	private boolean running = false;

	/**
	 * startupNode, if true, the node is not waiting for input data but executes
	 * the compute() method immediately
	 */
	private boolean startupNode = false;

	/**
	 * nodeName
	 */
	private String nodeName;

	/**
	 * continuous
	 */
	private boolean continuous;

	/**
	 * network
	 */
	private Network network;

	/**
	 * nodeThread the currently running thread
	 */
	private Thread nodeThread = null;

	/**
	 * Default AbstractNode constructor. Every subclass of AbstractNode should
	 * implement a default constructor for bean instantiation
	 * 
	 */
	public AbstractNode() {
		this("Unnamed AbstractNode " + autoNodeNr++, false);
	}

	/**
	 * AbstractNode Constructor - This is the "preferred" constructor for Nodes.
	 * Every sub-class should provide this constructor and call the AbstractNode
	 * class' super constructor
	 * 
	 * @param name
	 *            the (network unique) name of the node
	 * @param continuous
	 *            - if true, the node thread does not terminate until the node
	 *            is explicitly terminating. if false, the node only waits for
	 *            sufficient input ONCE, then computes and terminates
	 *            automatically
	 */
	public AbstractNode(String nodeName, boolean continuous) {
		super();
		setNodeName(nodeName);
		setContinuous(continuous);
	}

	/**
	 * createPorts
	 * 
	 */
	protected abstract void createPorts();

	/**
	 * startNode
	 * 
	 */
	public synchronized Thread startNode() {
		if (nodeThread == null) {
			nodeThread = new Thread(network.getThreadGroup(), this, getClass().getName());
			nodeThread.start();
		}
		return nodeThread;
	}

	/**
	 * @see java.lang.Runnable#run()
	 */
	public void run() {
		running = true;
		synchronized (this) {
			try {
				LOG_NODE.fine("+++ NODE " + getNodeName() + " started.");

				// node loop runs once for non-continuous nodes. for continuous
				// nodes, the value of the finished attribute is the termination
				// criteria
				do {

					// wait for data to arrive on required input ports
					while (!nodeReady()) {
						LOG_NODE.finer("AbstractNode " + getNodeName() + " waiting for data");
						try {
							Thread.sleep(NODE_READY_WAIT_DELAY);
						} catch (InterruptedException e) {
							LOG_NODE.warning(getNodeName()
									+ " interrupted while waiting for data on input ports: "
									+ e.getLocalizedMessage());
						}
					}

					// data should be available now. start the computation
					LOG_NODE.finest(getNodeName() + " running compute()");
					try {
						compute();
					} catch (ClassCastException cce) {
						LOG_NODE.severe(getNodeName() + ": Invalid data type: " + cce.getLocalizedMessage());
					} catch (Throwable t) {
						LOG_NODE.throwing(getNodeName(), "compute()", t);
					}

				} while (continuous && !isFinished());
			} finally {
				running = false;
			}
			LOG_NODE.fine("--- NODE " + getNodeName() + " finished.");
		}
	}

	/**
	 * @return true, if the node execution loop should terminate (for continuous
	 *         nodes). This method should be overridden by "continuous" nodes.
	 *         By default, this method returns "false" as long as there is data
	 *         on at least one input port. This also checks all
	 *         ConfigurationPorts.
	 */
	public boolean isFinished() {
		for (InputPort ip : getInputPorts()) {
			if (ip.hasDataElement()) {
				return false;
			}
		}
		return true;
	}

	/**
	 * @return true, if all required input nodes have dataall input nodes have
	 *         data available
	 */
	public synchronized boolean nodeReady() {
		// we are ready if there ARE NO input ports for which we could wait
		if (inputPortMap.size() == 0) {
			LOG_NODE.finest(getNodeName() + " ready (no ports)");
			return true;
		}

		// even if there are input ports, consider ourselves ready if this is a
		// "startup node".
		if (isStartupNode()) {
			// in this case, we drop the "startup" flag as we only want to start
			// this node without any further conditions ONCE
			setStartupNode(false);
			LOG_NODE.finest(getNodeName() + " ready (startup)");
			return true;
		}

		// check ALL required input ports for filled buffers
		for (InputPort ip : inputPortMap.values()) {
			LOG_NODE.finest(getNodeName() + "[" + ip.getName() + "]: REQ=" + ip.isRequired() + ", HASDATA="
					+ ip.hasData());
			if (ip.isRequired() && !ip.hasData()) {
				// at least one required input port doesn't contain data..
				return false;
			}
		}

		// we're still here? ok, now we can check if ANY of the nodes actually
		// has data
		for (InputPort ip : inputPortMap.values()) {
			if (ip.hasData()) {
				// there is data available..
				LOG_NODE.finest(getNodeName() + " ready (data available)");
				return true;
			}
		}

		// nope, nothing here for us
		return false;
	}

	/**
	 * the compute method should collect all data from the input ports (which
	 * are assumed to contain data at the time this method is called), computes
	 * the output and pushes the data into the output ports
	 */
	public abstract void compute();

	/**
	 * addInputPort
	 * 
	 * @param ip
	 * @return
	 */
	public InputPort addInputPort(InputPort ip) {
		if (inputPortMap.containsKey(ip.getName())) {
			throw new IllegalArgumentException("InputPort [" + ip.getName() + "] already present in "
					+ getNodeName());
		}
		inputPortMap.put(ip.getName(), ip);
		return ip;
	}

	/**
	 * addInputPort
	 * 
	 * @param name
	 * @param type
	 * @param required
	 * @return
	 */
	public InputPort addInputPort(String name, Class<?> type, boolean required) {
		return addInputPort(new InputPort(name, this, required, type));
	}

	/**
	 * Add an existing ConfigurationPort object to this node instance
	 * 
	 * @param cp
	 *            the ConfigurationPort to add
	 * @return the added ConfigurationPort
	 */
	public ConfigurationPort addConfigurationPort(ConfigurationPort cp) {
		return (ConfigurationPort) addInputPort(cp);
	}

	/**
	 * Add a new configuration port with name, type and default value to this
	 * node
	 * 
	 * @param name
	 *            the port name (Java variable naming convention)
	 * @param type
	 *            the class type of the expected data on this port
	 * @param defaultValue
	 *            a default value
	 * @return the new ConfigurationPort instance
	 */
	public ConfigurationPort addConfigurationPort(String name, Class<?> type, Object defaultValue) {
		return addConfigurationPort(new ConfigurationPort(name, this, true, type, defaultValue));
	}

	/**
	 * addConfigurationPort
	 * 
	 * @param name
	 * @param type
	 * @return
	 */
	public ConfigurationPort addConfigurationPort(String name, Class<?> type) {
		return addConfigurationPort(new ConfigurationPort(name, this, type));
	}

	/**
	 * configure
	 * 
	 * @param portName
	 * @param value
	 */
	public void configure(String portName, Object value) {
		InputPort ip = inputPortMap.get(portName);
		if (ip == null) {
			throw new IllegalArgumentException(getNodeName() + ": No such InputPort: " + portName);
		}
		if (ip instanceof ConfigurationPort) {
			ConfigurationPort cp = (ConfigurationPort) ip;
			if (value == null || cp.getType().isAssignableFrom(value.getClass())) {
				cp.setData(value);
			} else {
				throw new IllegalArgumentException(getNodeName()
						+ ": Invalid data type for ConfigurationPort " + cp.getName() + ". Expected "
						+ cp.getType().getName() + ", got " + value.getClass().getName());
			}
		} else {
			throw new IllegalStateException(getNodeName() + ": " + ip.getName()
					+ " is not a ConfigurationPort. Run-time data must be provided.");
		}
	}

	/**
	 * addOutputPort
	 * 
	 * @param op
	 * @return
	 */
	public OutputPort addOutputPort(OutputPort op) {
		if (outputPortMap.containsKey(op.getName())) {
			throw new IllegalArgumentException("OutputPort [" + op.getName() + "] already present in "
					+ getNodeName());
		}
		outputPortMap.put(op.getName(), op);
		return op;
	}

	/**
	 * addOutputPort
	 * 
	 * @param name
	 * @param type
	 * @return
	 */
	public OutputPort addOutputPort(String name, Class<?> type) {
		return addOutputPort(new OutputPort(name, this, type));
	}

	/**
	 * replaceInputPort - replaces a (possibly existing) input port. if the
	 * given port doesn't yet exist, the new port is simply added to this node
	 * 
	 * @param portName
	 * @param newInputPort
	 * @return
	 */
	public InputPort replaceInputPort(String portName, InputPort newInputPort) {
		InputPort oldPort = inputPortMap.remove(portName);
		inputPortMap.put(portName, newInputPort);
		if (oldPort == null) {
			LOG_NODE.fine("No such InputPort: " + portName + ". Adding new port without replacement.");
		} else {
			LOG_NODE.fine("InputPort " + portName + " has been replaced.");
		}
		return newInputPort;
	}

	/**
	 * replaceInputPort
	 * 
	 * @param newInputPort
	 * @return
	 */
	public InputPort replaceInputPort(InputPort newInputPort) {
		return replaceInputPort(newInputPort.getName(), newInputPort);
	}

	/**
	 * replaceOutputPort - replaces a (possibly existing) output port. if the
	 * given port doesn't yet exist, the new port is simply added to this node.
	 * 
	 * @param portName
	 * @param newOutputPort
	 * @return
	 */
	public OutputPort replaceOutputPort(String portName, OutputPort newOutputPort) {
		OutputPort oldPort = outputPortMap.remove(portName);
		outputPortMap.put(portName, newOutputPort);
		if (oldPort == null) {
			LOG_NODE.fine("No such OutputPort: " + portName + ". Adding new port without replacement.");
		} else {
			LOG_NODE.fine("OutputPort " + portName + " has been replaced.");
		}
		return newOutputPort;
	}

	/**
	 * replaceOutputPort
	 * 
	 * @param newOutputPort
	 * @return
	 */
	public OutputPort replaceOutputPort(OutputPort newOutputPort) {
		return replaceOutputPort(newOutputPort.getName(), newOutputPort);
	}

	/**
	 * getInputPort
	 * 
	 * @param name
	 * @return
	 */
	public InputPort getInputPort(String name) {
		InputPort ip = inputPortMap.get(name);
		if (ip == null) {
			throw new IllegalArgumentException("InputPort [" + name + "] not found in " + getNodeName());
		}
		return ip;
	}

	/**
	 * getInputPort - convenience method to get a node's ONLY input port
	 * 
	 * @return
	 */
	public InputPort getInputPort() {
		if (countDataInputPorts() != 1) {
			throw new IllegalStateException(getNodeName() + " has more than one input port.");
		}
		return getDataInputPort();
	}

	/**
	 * getDataInputPort
	 * 
	 * @return
	 */
	private InputPort getDataInputPort() {
		for (InputPort i : inputPortMap.values()) {
			if (!(i instanceof ConfigurationPort)) {
				return i;
			}
		}
		throw new IllegalStateException(getNodeName() + ": " + "No non-config InputPort available.");
	}

	/**
	 * countDataInputPorts returns the number of input ports, which are NOT
	 * configuration ports
	 * 
	 * @return
	 */
	private int countDataInputPorts() {
		int n = 0;
		for (InputPort i : inputPortMap.values()) {
			if (!(i instanceof ConfigurationPort)) {
				n++;
			}
		}
		return n;
	}

	/**
	 * getOutputPort
	 * 
	 * @param name
	 * @return
	 */
	public OutputPort getOutputPort(String name) {
		OutputPort op = outputPortMap.get(name);
		if (op == null) {
			throw new IllegalArgumentException("Output port '" + name + "' not found in node "
					+ getNodeName());
		}
		return op;
	}

	/**
	 * getOutputPort - convenience method to get a node's ONLY output port
	 * 
	 * @return
	 */
	public OutputPort getOutputPort() {
		if (outputPortMap.size() != 1) {
			throw new IllegalStateException(getNodeName() + " has more than one output port.");
		}
		return outputPortMap.values().iterator().next();
	}

	/**
	 * Convenience method for passthrough-nodes, which allows to forward a
	 * DataElement received on one port to another port, while retrieving the
	 * contained object for further processing.
	 * 
	 * @param ip
	 * @param op
	 * @return
	 */
	public Object fetchAndForwardData(InputPort ip, OutputPort op) {
		DataElement de = ip.fetchBuffer();
		op.pushBuffer(de);
		return de.getData();
	}

	/**
	 * @return the startupNode
	 */
	public boolean isStartupNode() {
		return startupNode;
	}

	/**
	 * @param startupNode
	 *            the startupNode to set
	 */
	public void setStartupNode(boolean startupNode) {
		this.startupNode = startupNode;
	}

	/**
	 * makeStartupNode
	 * 
	 * @return
	 */
	public AbstractNode makeStartupNode() {
		setStartupNode(true);
		return this;
	}

	/**
	 * addToNetwork
	 * 
	 * @param network
	 * @return
	 */
	public AbstractNode addToNetwork(Network network) {
		network.addNode(this);
		return this;
	}

	/**
	 * getInputPorts
	 * 
	 * @return
	 */
	public Collection<InputPort> getInputPorts() {
		return inputPortMap.values();
	}

	/**
	 * getOutputPorts
	 * 
	 * @return
	 */
	public Collection<OutputPort> getOutputPorts() {
		return outputPortMap.values();
	}

	/**
	 * @return the name
	 */
	public String getNodeName() {
		return nodeName;
	}

	/**
	 * @param name
	 *            the name to set
	 */
	public void setNodeName(String name) {
		this.nodeName = name;
	}

	/**
	 * @return the continuous
	 */
	public boolean isContinuous() {
		return continuous;
	}

	/**
	 * @param continuous
	 *            the continuous to set
	 */
	public void setContinuous(boolean continuous) {
		this.continuous = continuous;
	}

	/**
	 * @return the running
	 */
	public boolean isRunning() {
		return running;
	}

	/**
	 * @param running
	 *            the running to set
	 */
	public void setRunning(boolean running) {
		this.running = running;
	}

	/**
	 * @return the inputPorts
	 */
	public Map<String, InputPort> getInputPortMap() {
		return inputPortMap;
	}

	/**
	 * @param inputPorts
	 *            the inputPorts to set
	 */
	public void setInputPortMap(Map<String, InputPort> inputPortMap) {
		this.inputPortMap = inputPortMap;
	}

	/**
	 * @return the outputPorts
	 */
	public Map<String, OutputPort> getOutputPortMap() {
		return outputPortMap;
	}

	/**
	 * @param outputPorts
	 *            the outputPorts to set
	 */
	public void setOutputPortMap(Map<String, OutputPort> outputPortMap) {
		this.outputPortMap = outputPortMap;
	}

	/**
	 * @return the network
	 */
	public Network getNetwork() {
		return network;
	}

	/**
	 * @param network
	 *            the network to set
	 */
	public void setNetwork(Network network) {
		this.network = network;
	}

	/**
	 * setup the node immediately after the constructor has been called. this
	 * allows custom subclasses of Node to perform an early customisation based
	 * on any given string (e.g. XML data or key/value pairs)
	 * 
	 * @param setupData
	 *            the data to be used for setup purposes. specific to custom
	 *            node implementations.
	 */
	public void setup(Object setupData) {
		// nothing to do by default. subclasses must provide this functionality.
	}

}
